package com.lagou.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kudu.client.*;

import java.util.Map;

public class MySinkToKudu extends RichSinkFunction<Map<String, Object>> {
    private String masterAddr;
    private String tableName;

    KuduClient client;
    KuduSession kuduSession;
    KuduTable kuduTable;


    public MySinkToKudu(String masterAddr, String tableName) {
        this.masterAddr = masterAddr;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new KuduClient.KuduClientBuilder(masterAddr).build();
        kuduSession = client.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
        kuduTable = client.openTable(tableName);
//        super.open(parameters);
    }

    public void invoke(Map<String, Object> value, Context context) throws Exception {
        Insert insert = kuduTable.newInsert();
        PartialRow row = insert.getRow();
        row.addString("id", value.get("id").toString());
        row.addString("name", value.get("name").toString());
        row.addInt("age", Integer.valueOf(value.get("age").toString()));

        kuduSession.flush();
        kuduSession.apply(insert);
        kuduSession.close();
    }

    @Override
    public void close() throws Exception {
        client.close();
    }
}
